package com.atlocal.fun;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Field;
import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Struct;
import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.source.SourceRecord;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.hadoop.util.Time;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @ClassName MyDebeziumDeserializationSchema
 * @Description 自定义Oracle-CDC 序列化器
 * @Author kongjiangjiang
 * @Date 2023-04-27 23:25
 * @Version 1.0
 **/
@SuppressWarnings("all")
public class MyDebeziumDeserializationSchema implements DebeziumDeserializationSchema<String> {
    String[] args;
    Map map = new HashMap<String, String>();
    int i = 0;

    public MyDebeziumDeserializationSchema(String[] args) {
        this.args = args;
        ParameterTool parameter = ParameterTool.fromArgs(args);
        String tableInfo = parameter.get("tableInfo").trim();
        String[] splits = tableInfo.split("#");

        for (int k = 0; k < splits.length; k++) {
            String[] pkInfo = splits[k].trim().split("&");
            JSONObject json = new JSONObject();
            String[] keys = pkInfo[1].trim().split("@");
            for (int j = 0; j < keys.length; j++) {
                json.put(keys[j] + "", "");
            }

            map.put(pkInfo[0].split("\\.")[1], json);
        }
    }

    /**
     * @param sourceRecord 将接受到的数据封装为对象
     * @param collector    将数据处理后，向下游传递
     * @throws Exception
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        Struct valueStruct = (Struct) sourceRecord.value();
        Struct sourceStruct = valueStruct.getStruct("source");
        //获取数据库名称
        String database = sourceStruct.getString("db").toLowerCase();
        //获取表名
        String table = sourceStruct.getString("table").toLowerCase();
        //获取操作类型
        String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();
        if ("create".equals(type)) {
            type = "I";
        } else if ("insert".equals(type)) {
            type = "I";
        } else if ("update".equals(type)) {
            type = "U";
        } else if ("delete".equals(type)) {
            type = "D";
        } else if ("read".equals(type)) {
            type = "I";
        }


        //创建Json对象，封装采集到数据
        JSONObject jsonObj = new JSONObject();
        jsonObj.put("tableOwner", sourceStruct.getString("schema") + "");
        jsonObj.put("objnNo", i++ + "");
        jsonObj.put("operType", type);
        jsonObj.put("tableName", table);
        jsonObj.put("DSG_SCN", sourceStruct.getString("scn") + "");
        jsonObj.put("DSG_LOADERTIME", DateUtil.now() + "");
        jsonObj.put("DSG_TRANID", sourceStruct.getInt64("ts_ms") + "");
        jsonObj.put("DSG_SEQID", "test");
        //after=Struct{id=1,name=zs,age=18},
        Struct afterStruct = valueStruct.getStruct("after");
        JSONObject dataJsonObj = new JSONObject();
        if (afterStruct != null) {
            List<Field> fieldsArr = afterStruct.schema().fields();
            for (Field field : fieldsArr) {
                String name = field.name();
                Object value = afterStruct.get(name);
                //字段转换成小写
                dataJsonObj.put(name.toLowerCase(), value + "");
            }
        }
        jsonObj.put("columnInfo", dataJsonObj);
        jsonObj.put("PK", map.get(table));

        //发送数据
        collector.collect(jsonObj.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return TypeInformation.of(String.class);
    }
}
